package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        // 接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        // 聚合流式处理
        // 初始方法，返回值是消息的value
        // 真正的聚合操作，返回值是消息的value
        stream.map((key, value) -> {
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    // 重置消息的key:1234343434   和  value: likes:1
                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd());
                })
                //按照文章id进行聚合
                .groupBy((key, value) -> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                // 自行的完成聚合的计算
                .aggregate(() -> "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0", // 初始值
                        (key, value, aggValue) -> {
                            if (StringUtils.isBlank(value)) return aggValue; // value 为空, 返回初始值
                            String[] aggAry = aggValue.split(",");
                            int col = 0, com = 0, lik = 0, vie = 0;
                            for (String agg : aggAry) {
                                String[] _split = agg.split(":");
                                // 获得初始值，也是时间窗口内计算之后的值
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(_split[0])) {
                                    case COLLECTION:
                                        col = Integer.parseInt(_split[1]);
                                        break;
                                    case COMMENT:
                                        com = Integer.parseInt(_split[1]);
                                        break;
                                    case LIKES:
                                        lik = Integer.parseInt(_split[1]);
                                        break;
                                    case VIEWS:
                                        vie = Integer.parseInt(_split[1]);
                                        break;
                                }
                            }
                            // 累加操作
                            String[] valAry = value.split(":");
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])) {
                                case COLLECTION:
                                    col += Integer.parseInt(valAry[1]);
                                    break;
                                case COMMENT:
                                    com += Integer.parseInt(valAry[1]);
                                    break;
                                case LIKES:
                                    lik += Integer.parseInt(valAry[1]);
                                    break;
                                case VIEWS:
                                    vie += Integer.parseInt(valAry[1]);
                                    break;
                            }

                            String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                            log.debug("文章的id: {}", key);
                            log.debug("当前时间窗口内的消息处理结果：{}", formatStr);
                            return formatStr;
                        }, Materialized.as("hot-article-stream-count-001"))
                .toStream()
                // 转换消息的value
                .map((key, value) -> new KeyValue<>(key.key(), formatObj(key.key(), value)))
                // 发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;
    }

    /**
     * 格式化消息的value数据
     *
     * @param articleId 文章id
     * @param value     消息的value
     * @return 格式化之后的消息的value
     */
    public String formatObj(String articleId, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));
        // 格式:  COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为:{}", JSON.toJSONString(mess));
        return JSON.toJSONString(mess);

    }
}